package com.example.inboxdemo.component;

import com.example.inboxdemo.pojo.entity.PublishDTO;
import com.example.inboxdemo.rewrite.SseEmitterUTF8;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RTopic;
import org.redisson.api.RedissonClient;
import org.springframework.stereotype.Component;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.function.Consumer;

@Slf4j
@Component
@RequiredArgsConstructor
public class SseEmitterServer {
    public final static String INBOX_USER_TOPIC_PREFIX = "inbox_topic_user_";

    private final RedissonClient redissonClient;

    public static Map<String, SseEmitter> onlineUserEmitters = new HashMap<>(100);

    private final InboxUserTopicListener inboxTopicUserMessageListener;

    /**
     * 创建用户连接并返回 SseEmitter
     *
     * @return SseEmitter
     */
    public SseEmitter connect(String userId) throws IOException {
        // 为方便演示，假如同一个用户多终端同时在线，仅保留最后一个终端
        if (onlineUserEmitters.containsKey(userId)) {
            SseEmitter emitter = onlineUserEmitters.remove(userId);
            try {
                emitter.send("您已被顶下线...");
            } catch (IOException e) {
                log.info("该用户可能已下线");
            } finally {
                emitter.complete();
            }
        }
        // 设置超时时间(和token有效期一致，超时后不再推送消息)，0表示不过期。默认30秒，超过时间未完成会抛出异常：AsyncRequestTimeoutException
        SseEmitterUTF8 sseEmitter = new SseEmitterUTF8(0L);
        // 注册回调
        sseEmitter.onCompletion(completionCallBack(userId));
        sseEmitter.onError(errorCallBack(userId));
        sseEmitter.onTimeout(timeoutCallBack(userId));
        onlineUserEmitters.put(userId, sseEmitter);

        // 订阅个人站内消息
        RTopic topic = redissonClient.getTopic(INBOX_USER_TOPIC_PREFIX + userId);
        // 仅首次订阅时需要增加监听器，否则容器内会存在多个监听器，一条消息用户会收到数次
        if (topic.countListeners() == 0) {
            topic.addListener(PublishDTO.class, inboxTopicUserMessageListener);
        }
        sseEmitter.send(String.format("%s号用户，欢迎登录！", userId));
        return sseEmitter;
    }

    private Runnable completionCallBack(String userId) {
        return () -> {
            log.info("用户 {} 连接结束", userId);
            onlineUserEmitters.remove(userId);
        };
    }

    private Runnable timeoutCallBack(String userId) {
        return () -> {
            log.info("用户 {} 连接超时", userId);
            onlineUserEmitters.remove(userId);
        };
    }

    private Consumer<Throwable> errorCallBack(String userId) {
        return throwable -> {
            log.info("用户 {} 连接异常", userId);
            onlineUserEmitters.remove(userId);
        };
    }

}

